package cn.doitedu.api;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class _01_WordCount_V2 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("hdfs://doitedu:8020/doit46_ckpt");

        // 添加数据源
        DataStreamSource<String> stream = env.socketTextStream("doitedu", 8866);


        // 调各类算子，来进行运算逻辑的组装
        stream.map(String::toUpperCase)
                // 压平打散
                .flatMap((FlatMapFunction<String, String>) (value, out) -> {
                    String[] split = value.split(",");
                    for (String s : split) {
                        out.collect(s);
                    }
                }).returns(String.class)  // 显式指定函数返回值类型
                .filter(v-> !v.startsWith("a")) // 过滤数据
                .map(w-> Tuple2.of(w,1)).returns(new TypeHint<Tuple2<String, Integer>>() {})
                .keyBy(t->t.f0)
                .sum("f1")
                .print();

        env.execute("deepsea");

    }

}
